package com.talent.service.iot.mqtt.handler;

import com.talent.service.iot.mqtt.config.Constant;
import com.talent.service.iot.mqtt.config.TlAuthProperties;
import com.talent.service.iot.mqtt.model.TlMqttSession;
import com.talent.service.iot.mqtt.model.fix.TlMqttFixedHead;
import com.talent.service.iot.mqtt.model.payload.TlMqttConnectPayload;
import com.talent.service.iot.mqtt.model.request.TlMqttConnectReq;
import com.talent.service.iot.mqtt.model.request.TlMqttPublishReq;
import com.talent.service.iot.mqtt.model.response.TlMqttConnack;
import com.talent.service.iot.mqtt.model.variable.TlMqttConnectVariableHead;
import com.talent.service.iot.mqtt.service.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
public class TlMqttConnectHandler extends SimpleChannelInboundHandler<TlMqttConnectReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttConnectHandler.class);



    public final IMqttSubscriptionService subscriptionService;

    private final IMqttSessionService sessionService;

    private final IMqttRetainMsgService retainMsgService;

    private final IMqttMsgService msgService;

    private final TlAuthProperties authProperties;

    private final IAuthenticationService authenticationService;

    public TlMqttConnectHandler(IMqttSubscriptionService subscriptionService,
                                IMqttSessionService sessionService,
                                IMqttRetainMsgService retainMsgService,
                                IMqttMsgService msgService,
                                IAuthenticationService authenticationService,
                                TlAuthProperties authProperties){
        this.subscriptionService = subscriptionService;
        this.sessionService = sessionService;
        this.retainMsgService = retainMsgService;
        this.msgService = msgService;
        this.authenticationService = authenticationService;
        this.authProperties = authProperties;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttConnectReq req) throws Exception {
        logger.debug("in【CONNECT】 handler");
        Channel channel = ctx.channel();
        TlMqttConnectVariableHead variableHead = req.getVariableHead();
        TlMqttConnectPayload payload = req.getPayload();
        TlMqttFixedHead fixedHead = req.getFixedHead();
        String clientId = payload.getClientId();
        //是否清除会话
        int cleanSession = variableHead.getCleanSession();
        //连接的qos登陆
        MqttQoS mqttQoS = fixedHead.getQos();
        //是否有遗嘱消息
        int willFlag = variableHead.getWillFlag();
        //遗嘱消息是否是保留消息
        int willRetain = variableHead.getWillRetain();

        if (authProperties.isEnabled()) {
            if (!authenticationService.authenticationByUserName("","")) {
                return;
            }
        }
        boolean sessionPresent = cleanSession != 0;
        //如果是1的话 就重新清除之前的会话
        if(sessionPresent){
            actionOnCleanSession(clientId);
        }

        sessionService.find(clientId)
                        .doOnSuccess(e->{
                            logger.debug("client【{}】 connect",clientId);
                              if(sessionPresent || e==null) {
                                  logger.info("cleanSession is 1 or session is null build new session");
                                  e = TlMqttSession.of(clientId, channel, cleanSession);
                              }else{
                                  e.setCleanSession(cleanSession);
                                  e.setChannel(channel);
                              }
                            // 主要是客户端标识以及正常断开连接的标识位为false
                            channel.attr(AttributeKey.valueOf(Constant.CLIENT_ID)).set(clientId);
                            //断开标志位设备false，表示没有走正常断开的逻辑
                            channel.attr(AttributeKey.valueOf(Constant.DISCONNECT)).set(false);

                            TlMqttConnack res = TlMqttConnack.of(cleanSession, MqttConnectReturnCode.CONNECTION_ACCEPTED);
                            channel.writeAndFlush(res).addListener(future -> {});
                            logger.debug("send ack message to client 【{}】 connect  success",clientId);

                            //保存遗嘱消息
                            if (willFlag == 1) {
                                TlMqttPublishReq pugMsg = TlMqttPublishReq.of(payload.getWillTopic(),
                                        mqttQoS, willRetain!=0, payload.getWillMessage(),msgService.getMessageId());
                                e.setWillMessage(pugMsg);
                                if (willRetain !=0) {
                                    retainMsgService.save(payload.getWillTopic(), pugMsg).subscribe();
                                }
                            }

                            /*
                             * 如果CleanSession被设置为0
                             * ，服务器必须根据当前的会话状态恢复与客户端的通信（客户端的唯一标识作为会话的标识）。
                             * 如果没有与客户端唯一标识相关的绘画，服务端必须创建一个新的会话。
                             * 客户端和服务端在断开连接后必须存储会话[MQTT-3.1.2-4]
                             */

                            if(!sessionPresent){
                                logger.debug("save client【{}】 session",clientId);
                                sessionService.save(e);
                                republish(e).subscribe();
                            }

                            //添加心跳检测
                            Short heartbeat = variableHead.getKeepAlive();
                            ctx.pipeline().addLast(new IdleStateHandler(
                                    0, 0,  heartbeat, TimeUnit.SECONDS));

                        }).subscribe();

    }

    private Mono<String> startValidateConnect(TlMqttConnectVariableHead variableHead ,
                          TlMqttConnectPayload payload,
                          Channel channel) {
        Short protocolVersion = variableHead.getProtocolVersion();
        int cleanSession = variableHead.getCleanSession();

        if (!(protocolVersion.toString().equals(MqttVersion.MQTT_5.protocolLevel() + "") ||
                protocolVersion.toString().equals(MqttVersion.MQTT_3_1_1.protocolLevel() + ""))) {
            TlMqttConnack res = TlMqttConnack.of(cleanSession, MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
            channel.writeAndFlush(res);
            return Mono.error(new IllegalArgumentException("protocol Not Supported"));
        }
        String clientId = payload.getClientId();
        //客户端不允许接入
        if ("mqtt".equals(clientId)) {
            TlMqttConnack res = TlMqttConnack.of(cleanSession, MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED);
            channel.writeAndFlush(res);
            return Mono.error(new IllegalArgumentException("clientId no auth"));
        }

        //服务端不可用
        if (false) {
            TlMqttConnack res = TlMqttConnack.of(cleanSession, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
            channel.writeAndFlush(res);
            return Mono.error(new RuntimeException("协议不支持"));
        }


        //连接标识  如果连接标识不是0 则断开连接
        int reserved = variableHead.getReserved();
        if (reserved != 0) {
            logger.error("客户端【{}】的连接标识为【{}】，断开连接", clientId, reserved);
            return Mono.error(new RuntimeException("连接标识错误"));
        }
        logger.debug("validate over");
        return Mono.defer(()->Mono.just(clientId));
    }

    private void actionOnCleanSession(String clientId) {
        // sessionService.clear(String ClientId) 方法返回 true 则表明该 clientId 之前的 cleanSession = false, 那么应该继续清理
        // 订阅信息、pub 信息、 pubRel 信息, 否则无需清理
         sessionService.clear(clientId)
                .flatMap(e -> {
                    if (e) {
                        return Mono.when(
                                 subscriptionService.clear(clientId),
                                 msgService.clearPublishMessage(clientId),
                                 msgService.clearPubRelMessage(clientId)
                        );
                    } else {
                        return Mono.empty();
                    }
                }).subscribe();
    }


    /**
     * 上线时补发qos1与qos2的消息
     * @param session 会话
     * @return Void
     */
    private Mono<Void> republish(TlMqttSession session){
        //补发qos1与qos的2的消息 未回复的消息

        logger.debug("cleanSession is 0 and republish message");
        Set<Integer> publishMsgStore = session.getPublishMsgStore();

        for (Integer messageId : publishMsgStore) {
            logger.info("client【{}】 get reMessageId is【{}】",session.getClientId(),messageId);
            msgService.findPublishMessage(session.getClientId(), messageId)
                    .subscribe(publishReq -> session.getChannel().writeAndFlush(publishReq));
        }

        //补发qos2收到但是没有回复的消息
        Set<Integer> pubrelMsgStore = session.getPubrelMsgStore();
        for (Integer messageId : pubrelMsgStore) {
            msgService.findPubRelMessage(session.getClientId(), messageId)
                    .subscribe(publishReq -> session.getChannel().writeAndFlush(publishReq));
        }
        return Mono.empty();
    }
}
